package com.hzh.test01.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

// 使用FlinkSQl读取“input/data.csv”中数据，并输出为txt文件
// 根据类型编码type_code统计次数，并输出到output/data.csv中
public class FlinkSqlTest01 {
    public static void main(String[] args) {
        // TODO 1. 获取流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // TODO 2. 获取表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        // TODO 3. 读取CSV文件
        tableEnv.executeSql("CREATE TABLE dim_dic_code_import (code STRING, name STRING,chinese_name STRING  ,ename_name STRING  ,name_description STRING  ,parent_code STRING  ,type_code STRING  ,type_name STRING  ,data_source STRING  ,is_valid STRING  ,import_date String) with (" +
                "'connector' = 'filesystem'," +
                "'path' = 'input/data.csv'," +
                "'format' = 'csv')");
        //tableEnv.executeSql("select * from dim_dic_code_import").print();
        // TODO 4. 创建临时表
        tableEnv.executeSql("select type_code, count(1) from dim_dic_code_import group by type_code").print();


    }
}
